package com.eshop.storm.bolt;

import com.alibaba.fastjson.JSONObject;
import lombok.extern.slf4j.Slf4j;
import org.apache.storm.topology.BasicOutputCollector;
import org.apache.storm.topology.OutputFieldsDeclarer;
import org.apache.storm.topology.base.BaseBasicBolt;
import org.apache.storm.tuple.Fields;
import org.apache.storm.tuple.Tuple;
import org.apache.storm.tuple.Values;

/**
 * @author: Xukai
 * @description: 日志解析的bolt
 * @createDate: 2018/8/2 01:03
 * @modified By:
 */
@Slf4j
public class LogParseBolt extends BaseBasicBolt {

    private static final long serialVersionUID = -1877613187276886230L;

    @Override
    public void execute(Tuple input, BasicOutputCollector collector) {
        // 获取kafka发送的数据，是一个byte数组
        byte[] value = (byte[]) input.getValue(0);
        String message = new String(value);
        log.info("LogParseBolt get message: " + message);
        JSONObject messageJSON = JSONObject.parseObject(message);
        JSONObject uriArgsJSON = messageJSON.getJSONObject("uri_args");
        Long productId = uriArgsJSON.getLong("productId");
        if (productId != null) {
            collector.emit(new Values(productId));
        }
    }

    @Override
    public void declareOutputFields(OutputFieldsDeclarer declarer) {
        declarer.declare(new Fields("productId"));
    }
}
